package zookeeperdeploylogic

import (
	"context"
	"errors"
	"fmt"
	"github.com/zeromicro/go-zero/core/logx"
	"strings"
	"sync"
	"time"
	"yunzhan/common/models"
	utils "yunzhan/common/utils"
	"yunzhan/rpc-server/internal/svc"
	agent "yunzhan/rpc-server/pb"
)

type DeployZooKeeperLogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewDeployZooKeeperLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DeployZooKeeperLogic {
	return &DeployZooKeeperLogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

// DeployZooKeeper 部署ZooKeeper集群
func (l *DeployZooKeeperLogic) DeployZooKeeper(in *agent.DeployZookeeperRequest) (*agent.DeployResponse, error) {
	// todo: add your logic here and delete this line
	if in.NodeInfo == nil {
		return &agent.DeployResponse{
			Code:    500,
			Message: "未配置节点信息, 无法执行集群部署",
		}, errors.New("未配置节点信息, 无法执行集群部署")
	}

	// 立即返回一个响应，表示请求已接收并正在处理
	resp := &agent.DeployResponse{
		Code:    200,
		Message: fmt.Sprintf("请求已接收，正在部署 %s 到节点 %s", in.ConfigInfo.ComponentName, in.NodeInfo.Host),
	}

	go l.startDeployment(in)

	return resp, nil
}
func (l *DeployZooKeeperLogic) startDeployment(in *agent.DeployZookeeperRequest) {

	var wg sync.WaitGroup

	wg.Add(1)

	go func() {
		defer wg.Done()

		// 创建 LogManager 实例
		logManager := models.NewLogManager(100, fmt.Sprintf("http://%s:%d", l.svcCtx.Config.RestServices["apiServer"].RestConf.Host, l.svcCtx.Config.RestServices["apiServer"].RestConf.Port))
		defer logManager.Close()

		flushInterval := 5 * time.Second // 刷新间隔
		maxBatchSize := 100              // 最大批量大小
		retryDelay := 2 * time.Second    // 重试延迟
		maxRetries := 3                  // 最大重试次数

		// 创建 ComponentInfoManager 实例
		componentInfoManager := models.NewComponentInfoManager(
			1000,
			fmt.Sprintf("http://%s:%d", l.svcCtx.Config.RestServices["apiServer"].RestConf.Host, l.svcCtx.Config.RestServices["apiServer"].RestConf.Port),
			flushInterval,
			maxBatchSize,
			retryDelay,
			maxRetries)
		defer componentInfoManager.Close()

		packageDirectory := in.ConfigInfo.PackagePath
		rtrimDir := utils.Rtrim(packageDirectory, "/")
		zookeeperVersion := in.ConfigInfo.ComponentName
		installPath := utils.Rtrim(in.ConfigInfo.InstallPath, "/")

		baseInfo := fmt.Sprintf("安装包存放目录: %s, Zookeeper version: %s, Zookeeper 安装目录: %s", rtrimDir, zookeeperVersion, installPath)
		step := 1
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", baseInfo, "Zookeeper", "Server")

		javaHome := "/usr/local/java"

		// 检测安装路径是否存在
		checkPathCommand := fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
`, installPath, installPath)

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "执行组件安装路径检查...", "Zookeeper", "QuorumPeerMain")
		output, err := utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件安装路径异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件安装路径完成", "Zookeeper", "QuorumPeerMain")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "解压安装包...", "Zookeeper", "QuorumPeerMain")

		tarCommand := fmt.Sprintf("sudo tar -xzf %s/%s -C %s", rtrimDir, zookeeperVersion, installPath)
		output, err = utils.ExecCommand(tarCommand)
		if err != nil {
			message := fmt.Sprintf("解压安装包异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "解压安装包完成", "Zookeeper", "QuorumPeerMain")

		zookeeperHome := utils.Rtrim(zookeeperVersion, ".tar.gz")
		home := fmt.Sprintf("%s/%s", installPath, zookeeperHome)

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "修改组件所属用户组...", "Zookeeper", "QuorumPeerMain")
		chownCommand := fmt.Sprintf("sudo chown -R %s:%s %s", in.NodeInfo.Username, in.NodeInfo.Username, home)
		output, err = utils.ExecCommand(chownCommand)
		if err != nil {
			message := fmt.Sprintf("修改组件所属用户组异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "修改组件所属用户组完成", "Zookeeper", "QuorumPeerMain")

		checkPathCommand = fmt.Sprintf(`
if [ ! -d "%s" ]; then
sudo mkdir -p "%s"
fi
sudo chown -R %s:%s %s
`, in.ZookeeperConfig.DataDir, in.ZookeeperConfig.DataDir, in.NodeInfo.Username, in.NodeInfo.Username, in.ZookeeperConfig.DataDir)
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件数据存储目录...", "Zookeeper", "QuorumPeerMain")
		output, err = utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件数据存储目录异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件数据存储目录完成", "Zookeeper", "QuorumPeerMain")
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建zoo.cfg配置文件...", "Zookeeper", "QuorumPeerMain")
		cpCommand := fmt.Sprintf("cp -r %s/conf/zoo_sample.cfg %s/conf/zoo.cfg", home, home)
		output, err = utils.ExecCommand(cpCommand)
		if err != nil {
			message := fmt.Sprintf("创建zoo.cfg配置文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建zoo.cfg配置文件完成", "Zookeeper", "QuorumPeerMain")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置zoo.cfg文件...", "Zookeeper", "QuorumPeerMain")
		dataDir := utils.Rtrim(in.ZookeeperConfig.DataDir, "/")
		// 修改配置文件zoo.cfg
		configCommand := fmt.Sprintf("sed -i '/^\\s*#\\?\\s*tickTime=/c\\tickTime=%s' %s/conf/zoo.cfg", in.ZookeeperConfig.TickTime, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*dataDir=/c\\dataDir=%s' %s/conf/zoo.cfg", dataDir, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*clientPort=/c\\clientPort=%s' %s/conf/zoo.cfg", in.ZookeeperConfig.ClientPort, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*initLimit=/c\\initLimit=%s' %s/conf/zoo.cfg", in.ZookeeperConfig.InitLimit, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*syncLimit=/c\\syncLimit=%s' %s/conf/zoo.cfg", in.ZookeeperConfig.SyncLimit, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*maxClientCnxns=/c\\maxClientCnxns=%s' %s/conf/zoo.cfg", in.ZookeeperConfig.MaxClientCnxns, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*autopurge.snapRetainCount=/c\\autopurge.snapRetainCount=%s' %s/conf/zoo.cfg", in.ZookeeperConfig.AutopurgeSnapRetainCount, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*autopurge.purgeInterval=/c\\autopurge.purgeInterval=%s' %s/conf/zoo.cfg", in.ZookeeperConfig.AutopurgePurgeInterval, home)
		configCommand += fmt.Sprintf(" && sed -i '/^\\s*#\\?\\s*dataLogDir=/c\\dataLogDir=%s' %s/conf/zoo.cfg", home+"/logs", home)
		//logx.Infof("zoo.cfg配置修改命令: %s", configCommand)
		output, err = utils.ExecCommand(configCommand)
		if err != nil {
			message := fmt.Sprintf("配置zoo.cfg文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置zoo.cfg文件完成", "Zookeeper", "QuorumPeerMain")

		var serverConfigs []string
		for key, value := range in.ZookeeperConfig.ServerList {
			serverConfig := fmt.Sprintf("%s=%s", strings.TrimSpace(key), strings.TrimSpace(value))
			serverConfigs = append(serverConfigs, serverConfig)
		}
		allServerConfigs := strings.Join(serverConfigs, "\n")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "添加servers配置...", "Zookeeper", "QuorumPeerMain")
		addServerCommand := fmt.Sprintf("echo '%s' | sudo tee -a %s/conf/zoo.cfg", allServerConfigs, home)
		_, err = utils.ExecCommand(addServerCommand)
		if err != nil {
			message := fmt.Sprintf("添加servers配置异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "添加servers配置完成", "Zookeeper", "QuorumPeerMain")

		myId := in.ZookeeperConfig.MyId[in.NodeInfo.Host]
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建myid文件...", "Zookeeper", "QuorumPeerMain")
		myidCommand := fmt.Sprintf("echo %d > %s/myid", myId, utils.Rtrim(in.ZookeeperConfig.DataDir, "/"))
		_, err = utils.ExecCommand(myidCommand)
		if err != nil {
			message := fmt.Sprintf("创建myid文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建myid文件完成", "Zookeeper", "QuorumPeerMain")

		zookeeperServiceFile := fmt.Sprintf(`[Unit]
Description=Apache ZooKeeper Service
Documentation=https://zookeeper.apache.org/
After=network.target

[Service]
Type=forking
User=%s
Group=%s
Environment="JAVA_HOME=%s"
#ExecStart=/bin/zkServer.sh start /conf/zoo.cfg
#ExecStop=/bin/zkServer.sh stop
#ExecReload=/bin/zkServer.sh restart
ExecStart=/bin/bash -c "export JAVA_HOME=%s && %s/bin/zkServer.sh start %s/conf/zoo.cfg"
ExecStop=/bin/bash -c "export JAVA_HOME=%s && %s/bin/zkServer.sh stop"
ExecReload=/bin/bash -c "export JAVA_HOME=%s && %s/bin/zkServer.sh restart"
PIDFile=%s/zookeeper_server.pid
KillMode=none
Restart=on-failure
LimitNOFILE=65536

[Install]
WantedBy=multi-user.target`, in.NodeInfo.Username, in.NodeInfo.Username, javaHome, javaHome, home, home, javaHome, home, javaHome, home, dataDir)

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建zookeeper.service文件...", "Zookeeper", "QuorumPeerMain")
		writeServiceFileCommand := fmt.Sprintf("echo '%s' | sudo tee /etc/systemd/system/zookeeper.service", zookeeperServiceFile)
		_, err = utils.ExecCommand(writeServiceFileCommand)
		if err != nil {
			message := fmt.Sprintf("创建 zookeeper.service 文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建zookeeper.service文件完成", "Zookeeper", "QuorumPeerMain")
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动Zookeeper服务...", "Zookeeper", "QuorumPeerMain")
		reloadSystemdCommand := fmt.Sprintf("sudo systemctl daemon-reload && sudo systemctl enable zookeeper.service --now")
		output, err = utils.ExecCommand(reloadSystemdCommand)
		if err != nil {
			message := fmt.Sprintf("启动Zookeeper服务异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动Zookeeper服务完成", "Zookeeper", "QuorumPeerMain")

		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "等待Zookeeper服务完全启动", "Zookeeper", "QuorumPeerMain")

		time.Sleep(3 * time.Second)

		step++
		statusCommand := fmt.Sprintf("sudo systemctl is-active zookeeper.service")
		output, err = utils.ExecCommand(statusCommand)
		if err != nil || strings.TrimSpace(output) != "active" {
			message := fmt.Sprintf("Zookeeper启动后未正常运行: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "Zookeeper部署成功", "Zookeeper", "QuorumPeerMain")

		currentIP, currentIPErr := utils.GetCurrentInternalIP()
		if err != nil {
			message := fmt.Sprintf("%s: 无法获取当前主机IP: %v", in.NodeInfo.Host, currentIPErr)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}
		currentHost, currentHostErr := utils.GetCurrentHostname()
		if err != nil {
			message := fmt.Sprintf("%s: 无法获取当前主机名: %v", in.NodeInfo.Host, currentHostErr)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Zookeeper", "QuorumPeerMain")
			return
		}

		nodeRole, err := utils.GetZooKeeperRole(in.NodeInfo.Host + ":" + in.ZookeeperConfig.ClientPort)
		if err != nil {
			logx.Errorf("获取 ZooKeeper 角色信息失败: %v", err)
			return
		}
		zookeeperInfo := &models.ComponentInfo{
			ClusterID:      in.ClusterID,
			ComponentName:  "Zookeeper",
			Version:        in.Version,
			NodeHost:       currentHost,
			NodeIP:         currentIP,
			ComponentRole:  nodeRole,
			HomePath:       home,
			DataStorageDir: in.ZookeeperConfig.DataDir,
			Port:           in.ZookeeperConfig.ClientPort,
			Status:         true,
			AdditionalInfo: fmt.Sprintf("config_file=%s/conf/zoo.cfg", home),
		}

		componentInfoManager.SendComponentInfo(zookeeperInfo)

	}()
	wg.Wait()

}
